package com.navinfo.platform.trip.analysis.flink.source;

import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import com.navinfo.opentsp.platform.location.protocol.common.RealTimeDataPb;
import com.navinfo.platform.trip.analysis.flink.watermark.BoundedOutOfOrdernessWatermarks;
import com.navinfo.platform.trip.common.pojo.PointProtocol;
import com.navinfo.platform.trip.analysis.flink.config.SystemConfig;
import com.navinfo.platform.trip.common.util.PointUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.net.URISyntaxException;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * 读取Kafka数据生成DataStreamSource
 * @author: web
 * @data: 2020-04-07
 **/
@SuppressWarnings("unchecked")
public class KafkaConsumer implements Serializable {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    /**
     * 配置工具类
     */
    private ParameterTool parameterTool;

    /**
     * 位置点0200消费者的配置
     */
    private Properties locationProperties;
    /**
     * 密集0F37消费者的配置
     */
    private Properties realtimeProperties;

    /**
     * 位置点0200的topic
     */
    private String locationTopic;
    /**
     * 密集0F37的topic
     */
    private String realtimeTopic;

    public KafkaConsumer(String[] args) throws IOException, URISyntaxException {
        parameterTool = SystemConfig.loadJarPathConfig(args, "kafka_source.properties");

        locationProperties = new Properties();
        locationProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, parameterTool.get("kafka.consumer.location.servers"));
        locationProperties.put(ConsumerConfig.GROUP_ID_CONFIG, parameterTool.get("kafka.consumer.location.group.id", "trip_analysis"));
        locationProperties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.heartbeat.interval.ms",3000));
        locationProperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.session.timeout.ms",10000));
        locationProperties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.max.poll.interval.ms", 300000));
        locationProperties.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, this.parameterTool.getInt("kafka.consumer.fetch.min.bytes",1024));
        locationProperties.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.fetch.max.wait.ms",500));
        locationProperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, this.parameterTool.getInt("kafka.consumer.max.poll.records",500));
        locationProperties.put(ConsumerConfig.SEND_BUFFER_CONFIG,  this.parameterTool.getInt("kafka.consumer.send.buffer.bytes",131072));
        locationProperties.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG,  this.parameterTool.getInt("kafka.consumer.receive.buffer.bytes",65536));

        realtimeProperties = new Properties();
        realtimeProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, parameterTool.get("kafka.consumer.realtime.servers"));
        realtimeProperties.put(ConsumerConfig.GROUP_ID_CONFIG, parameterTool.get("kafka.consumer.realtime.group.id", "trip_analysis"));
        realtimeProperties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.heartbeat.interval.ms",3000));
        realtimeProperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.session.timeout.ms",10000));
        realtimeProperties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.max.poll.interval.ms", 300000));
        realtimeProperties.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, this.parameterTool.getInt("kafka.consumer.fetch.min.bytes",1024));
        realtimeProperties.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, this.parameterTool.getInt("kafka.consumer.fetch.max.wait.ms",500));
        realtimeProperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, this.parameterTool.getInt("kafka.consumer.max.poll.records",500));
        realtimeProperties.put(ConsumerConfig.SEND_BUFFER_CONFIG,  this.parameterTool.getInt("kafka.consumer.send.buffer.bytes",131072));
        realtimeProperties.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG,  this.parameterTool.getInt("kafka.consumer.receive.buffer.bytes",65536));

        //动态发现Topic或Partition consumerConf.put(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, Time.minutes(5).toMilliseconds());
        locationTopic = this.parameterTool.get("kafka.consumer.location.topic", "locationDataPB");
        realtimeTopic = this.parameterTool.get("kafka.consumer.realtime.topic", "concentratedDataPB");


        logger.info("初始化Kafka消费者完成...");
    }

    /**
     * 位置点数据源
     * @return 消费Kafka中的位置点数据
     */
    public FlinkKafkaConsumer<PointProtocol> locationData() {
        FlinkKafkaConsumer<PointProtocol> locationSource = new FlinkKafkaConsumer<>(locationTopic,
            new KafkaDeserializationSchema<PointProtocol>() {
                @Override
                public PointProtocol deserialize(ConsumerRecord<byte[], byte[]> record) {
                    PointProtocol pointProtocol = null;
                    try {
                        pointProtocol = PointUtils.toPointProtocol(LCLocationData.LocationData.parseFrom(record.value()));
                    }catch (Exception e){
                        String errMsg = "解析Kafka位置数据Topic[" + record.topic() +
                                "]，partition[" + record.partition() +
                                "]，offset[" + record.offset() + "]异常";
                        logger.error(errMsg, e);
                    }

                    //方法抛出异常会导致作业失败并重新启动，返回null跳过损坏的消息；
                    return pointProtocol;
                }

                @Override
                public TypeInformation<PointProtocol> getProducedType() {
                    return TypeInformation.of(PointProtocol.class);
                }
                @Override
                public boolean isEndOfStream(PointProtocol nextElement) {
                    return false;
                }
            }, locationProperties);

        //设置偏移量
        setStartOffsets(locationSource, 1);

        logger.info("创建Kafka主题{}的消费者完成...", locationTopic);
        return locationSource;
    }

    /**
     * 密集数据源
     * @return 消费kafka中的密集数据
     */
    public FlinkKafkaConsumer<PointProtocol> realTimeData(){
        FlinkKafkaConsumer<PointProtocol> realtimeSource = new FlinkKafkaConsumer<>(realtimeTopic, new KafkaDeserializationSchema<PointProtocol>() {
            @Override
            public PointProtocol deserialize(ConsumerRecord<byte[], byte[]> record) {
                PointProtocol pointProtocol = null;
                try {
                    pointProtocol = PointUtils.toPointProtocol(RealTimeDataPb.RealTimeData.parseFrom(record.value()));
                }catch (Exception e){
                    StringBuilder errMsg = new StringBuilder("解析Kafka密集数据Topic[").append(record.topic())
                            .append("]，partition[").append(record.partition())
                            .append("]，offset[").append(record.offset()).append("]异常");
                    logger.error(errMsg.toString(), e);
                }

                //方法抛出异常会导致作业失败并重新启动，返回null跳过损坏的消息；
                return pointProtocol;
            }

            @Override
            public TypeInformation<PointProtocol> getProducedType() {
                return TypeInformation.of(PointProtocol.class);
            }
            @Override
            public boolean isEndOfStream(PointProtocol nextElement) {
                return false;
            }
        }, realtimeProperties);

        //设置偏移量
        setStartOffsets(realtimeSource, 2);

        logger.info("创建Kafka主题{}的消费者完成...", realtimeTopic);
        return  realtimeSource;
    }

    /**
     * @return 位置数据topic的分区数
     */
    public int locationPartitionSize(){
        return parameterTool.getInt("kafka.consumer.location.partition.size", 8);
    }

    /**
     * @return 密集数据topic的分区数
     */
    public int realTimePartitionSize(){
        return parameterTool.getInt("kafka.consumer.realtime.partition.size", 8);
    }

    /**
     * 设置消费的起始位置
     * @param consumer kafka消费者
     * @param type  1位置数据，2密集数据
     */
    private void setStartOffsets(FlinkKafkaConsumer<PointProtocol> consumer, int type){
        String tmpStr = "";
        String tmpTopic = "";
        if(type == 1){
            tmpStr = "location";
            tmpTopic = locationTopic;
        }else if(type == 2){
            tmpStr = "realtime";
            tmpTopic = realtimeTopic;
        }

        int offsetType = parameterTool.getInt("kafka.consumer."+tmpStr+".offset.type", 0);

        //0:默认GroupOffsets，1:从头消费earliest，2:最新消费latest
        //3:指定timestamp，4:指定Group和OffSet，SpecificOffsets
        switch (offsetType){
            case 1 :
                consumer.setStartFromEarliest(); break;
            case 2 :
                consumer.setStartFromLatest(); break;
            case 3 :
                long time = parameterTool.getLong("kafka.consumer." + tmpStr + ".offset.3", System.currentTimeMillis());
                consumer.setStartFromTimestamp(time);
                break;
            case 4 :
                Map<KafkaTopicPartition, Long> specificStartupOffsets = new HashMap<>();
                //目前支持最多50个分区
                for(int i=0; i<50; i++){
                    String partitionOffset = parameterTool.get("kafka.consumer." + tmpStr + ".offset.4." + i) ;
                    if(null != partitionOffset){
                        specificStartupOffsets.put(new KafkaTopicPartition(tmpTopic, i), Long.valueOf(partitionOffset));
                    }else{
                        break;
                    }
                }
                consumer.setStartFromSpecificOffsets(specificStartupOffsets);
                break;
            default:
                consumer.setStartFromGroupOffsets(); break;
        }
    }
}
